package com.code.soulnetty.modbus.client;

import com.code.soulnetty.modbus.biz.ChannelInfo;
import com.code.soulnetty.modbus.client.channel.ChannelFactory;
import com.code.soulnetty.modbus.client.channel.impl.ProtocolChannel;
import com.code.soulnetty.modbus.client.channel.impl.ProtocolChannelCache;
import com.code.soulnetty.modbus.client.message.ProtocolRequest;
import com.code.soulnetty.modbus.client.message.ProtocolResponse;
import com.code.soulnetty.modbus.client.service.ProtocolResponseService;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.code.soulnetty.modbus.biz.Utils.*;

/**
 * netty 客户端封装，初始化netty连接，可以兼容不同协议的 channel 连接获取
 * @author lyf
 * @date 2022-12-16 23:55
 */
@Slf4j
public class NettyClient {

    final Bootstrap bootstrap;
    //
    final NettyConfig nettyConfig;
    final ChannelFactory channelFactory;

    private ProtocolChannel protocolChannel;
    private Function<NettyClient, ChannelHandler[]> channelHandlers;
    private ProtocolResponseService responseService;

    public NettyClient(NettyConfig nettyConfig, ChannelFactory channelFactory, Function<NettyClient, ChannelHandler[]> channelHandlers,
                       ProtocolResponseService responseService) {
        this.bootstrap = new Bootstrap();
        this.nettyConfig = nettyConfig;
        this.channelFactory = channelFactory;
        this.channelHandlers = channelHandlers;
        this.responseService = responseService;
        init();
    }

    private void init() {
        Class<? extends Channel> channelClass = getChannelClass();
        bootstrap.group(this.nettyConfig.getEventLoopGroup())
                .channel(channelClass)
                .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                .option(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        if (this.nettyConfig.getRcvBuf() > 0) {
            bootstrap.option(ChannelOption.SO_RCVBUF, this.nettyConfig.getRcvBuf());
        }
        if (this.nettyConfig.getSendBuf() > 0) {
            bootstrap.option(ChannelOption.SO_SNDBUF, this.nettyConfig.getSendBuf());
        }
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.nettyConfig.getConnectTimeout());
        if (this.nettyConfig.getEventLoopGroup() instanceof EpollEventLoopGroup) {
            bootstrap.handler(new MyEpollSocketChannelInitializer(channelHandlers, this));
        } else {
            bootstrap.handler(new MyNioChannelInitializer(channelHandlers, this));
        }
    }

    public ChannelFuture connect(InetSocketAddress address, int retryCount, long timeout, long reqIntl) {
        ChannelFuture channelFuture = bootstrap.connect(address);
        // 最长等待1s建立连接
        try {
            channelFuture.await(1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("CONN_ERR", e);
            throw new RuntimeException(e);
        }
        Throwable cause = channelFuture.cause();
        if (cause != null) {
            throw new RuntimeException(cause);
        }
        ProtocolChannel oldChannel1 = ProtocolChannelCache.get(address);
        if (oldChannel1 != null) {
            oldChannel1.close();
        }
        this.protocolChannel = channelFactory.create(address, channelFuture.channel(), responseService, retryCount, timeout, reqIntl);
        ProtocolChannelCache.put(this.protocolChannel);
        return channelFuture;
    }

    /**
     * 重连 目的就是连接成功，获取通道
     * @param channelInfo
     */
    public void reConn(ChannelInfo channelInfo) throws Exception {
        if (this.protocolChannel != null && this.protocolChannel.getRemoteAddress() instanceof InetSocketAddress) {
            connect((InetSocketAddress) this.protocolChannel.getRemoteAddress(), protocolChannel.getRetryCount(),
                    protocolChannel.getTimeout(), protocolChannel.getReqIntl());
        } else if (this.protocolChannel == null) {
            Map<String, String> props = channelInfo.getConfig();
            //重试次数
            int retryCount = MapUtils.getInteger(props, RETRY_COUNT_STR, DEFAULT_RETRY_COUNT);
            //轮询间隔
            long reqIntl = MapUtils.getLong(props, REQ_INTL, DEFAULT_REQ_INTL);
            //超时时间
            long timeOut = MapUtils.getLong(props, TIME_OUT, DEFAULT_TIME_OUT);
            connect(new InetSocketAddress(channelInfo.getIp(), channelInfo.getPort()), retryCount, timeOut, reqIntl);
        }
    }

    public CompletableFuture<ProtocolResponse> send(ProtocolRequest request) throws Exception {
        if (this.protocolChannel != null && this.protocolChannel.isConnected()) {
            return this.protocolChannel.send(request);
        } else {
            log.warn("protocol channel null or non connect");
            return null;
        }
    }

    public void shutdown() {
        ProtocolChannelCache.remove(this.protocolChannel.getRemoteAddress());
        this.protocolChannel.close();
    }

    private Class<? extends Channel> getChannelClass() {
        return this.nettyConfig.getEventLoopGroup() instanceof EpollEventLoopGroup ?
                EpollSocketChannel.class : NioSocketChannel.class;
    }

    /**
     * 获取远程服务地址
     *
     * @return 远程服务地址和端口
     */
    public InetSocketAddress getRemoteAddress() {
        if (protocolChannel == null) {
            return null;
        }
        return (InetSocketAddress) protocolChannel.getRemoteAddress();
    }

    /**
     * 判断当前是否链接
     *
     * @return true 已链接 ,false 未链接
     */
    public boolean isConnected() {
        return this.protocolChannel != null && this.protocolChannel.isConnected();
    }

    public static class MyNioChannelInitializer extends ChannelInitializer<NioSocketChannel> {
        private Function<NettyClient, ChannelHandler[]> channels;
        private NettyClient client;

        public MyNioChannelInitializer(Function<NettyClient, ChannelHandler[]> channels, NettyClient client) {
            this.channels = channels;
            this.client = client;
        }

        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
            ch.pipeline().addLast(channels.apply(client));
        }
    }

    public static class MyEpollSocketChannelInitializer extends ChannelInitializer<EpollSocketChannel> {
        private Function<NettyClient, ChannelHandler[]> channels;
        private NettyClient client;

        public MyEpollSocketChannelInitializer(Function<NettyClient, ChannelHandler[]> channels, NettyClient client) {
            this.channels = channels;
            this.client = client;
        }

        @Override
        protected void initChannel(EpollSocketChannel ch) throws Exception {
            ch.pipeline().addLast(channels.apply(client));
        }
    }
}
